{
 "nbformat": 4,
 "nbformat_minor": 2,
 "metadata": {
  "language_info": {
   "name": "python",
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "version": "3.8.1-final"
  },
  "orig_nbformat": 2,
  "file_extension": ".py",
  "mimetype": "text/x-python",
  "name": "python",
  "npconvert_exporter": "python",
  "pygments_lexer": "ipython3",
  "version": 3,
  "kernelspec": {
   "name": "python38164bitjuconda2d335684d9b14f4b9d4db7c47b222a54",
   "display_name": "Python 3.8.1 64-bit ('ju': conda)"
  }
 },
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 并发与异步（去阻塞）\n",
    "## 多线程\n",
    "在python 中多线程由PIL实现，既一个cup不停快速切换来模拟多个线程同时运行，由于线程切换额外的资源消耗python多线程有时并不能优化程序。当有IO或网络等耗时而无需消耗计算资源的任务时比较有效  \n",
    "使用`threading.Thread()`函数来创建新线程， `thread.start()` 来启动线程，`thread.join()`来等待线程执行完毕，通过`thread.setDaemon()`将其设定为守护线程。在线程间传输数据可以使用`queue.Queue`对象。  \n",
    "基本使用方法如下："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": "4\n124362309834884\n497448722055172\n1245084426971780\nmultil thread costs: 10.59255862236023\n4\n124362309834884\n497448722055172\n1245084426971780\nnormal costs: 11.178232669830322\n"
    }
   ],
   "source": [
    "import threading\n",
    "import time\n",
    "import logging\n",
    "\n",
    "num=10000000\n",
    "\n",
    "def f1(a):\n",
    "    res = 0\n",
    "    # print(a)\n",
    "    for i in range(a):\n",
    "        res += i ^ 2+3*i+2\n",
    "    print(res)\n",
    "\n",
    "\n",
    "def mutil_thread():\n",
    "    threads = []\n",
    "    for i in range(4):\n",
    "        t = threading.Thread(target=f1, args=(i*num+1,))\n",
    "        t.start()\n",
    "        threads.append(t)\n",
    "    for t in threads:\n",
    "        t.join()\n",
    "\n",
    "\n",
    "def normal():\n",
    "    for i in range(4):\n",
    "        f1(i*num+1)\n",
    "    # print(threading.activeCount())\n",
    "    # print(threading.enumerate())\n",
    "t1 = time.time()\n",
    "q1 = mutil_thread()\n",
    "print('multil thread costs:',time.time()-t1)\n",
    "\n",
    "t2 = time.time()\n",
    "normal()\n",
    "print('normal costs:',time.time()-t2)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "也可以继承`Thread`类并重写`run()`方法来创建线程，重写`_init_()`来自定义类，传递参数等。  \n",
    "注意不要覆盖除`_init_()`和`run()`之外的函数："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": "(Thread-18) running with (0,) and {'a': 1, 'b': 2}\n(Thread-19) running with (1,) and {'a': 1, 'b': 2}\n(Thread-20) running with (2,) and {'a': 1, 'b': 2}\n"
    }
   ],
   "source": [
    "logging.basicConfig(level=logging.DEBUG,\n",
    "                    format='(%(threadName)-9s) %(message)s',)\n",
    "\n",
    "class MyThread(threading.Thread):\n",
    "    def __init__(self, group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None):\n",
    "        super(MyThread,self).__init__(group=group, target=target, name=name, daemon=daemon)\n",
    "        self.args = args\n",
    "        self.kwargs = kwargs\n",
    "        return\n",
    "\n",
    "    def run(self):\n",
    "        logging.debug('running with %s and %s', self.args, self.kwargs)\n",
    "        return\n",
    "\n",
    "for i in range(3):\n",
    "    t = MyThread(args=(i,), kwargs={'a':1, 'b':2})\n",
    "    t.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "当需要让多个线程有序的穿插运行（如线程一计算得初步结果，线程二计算得中间结果，线程一再完成计算），可以使用锁  \n",
    "使用`threading.Lock()`或`threading.RLock`来创建锁（递归锁），`l.acquire()`来加锁，`l.release()`来释放，用法如下：\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": "1\n2\n3\n4\n5\n6\n7\n8\n9\n10\n20\n30\n40\n50\n60\n70\n80\n90\n100\n110\n"
    }
   ],
   "source": [
    "def job(i):\n",
    "    global a\n",
    "    global lock\n",
    "    lock.acquire()\n",
    "    for k in range(10):\n",
    "        a = a+i\n",
    "        print(a)\n",
    "    lock.release()\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    a = 0\n",
    "    lock=threading.Lock()\n",
    "    t1 = threading.Thread(target=job, args=(1,))\n",
    "    t2 = threading.Thread(target=job, args=(10,))\n",
    "    t1.start()\n",
    "    t2.start()\n",
    "    threading.RLock\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "也可使用条件对象：  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": "(consumer1) Consumer thread started ...\n(consumer1) Consumer waiting ...\n(consumer2) Consumer thread started ...\n(producer ) Producer thread started ...\n(consumer2) Consumer waiting ...\n(producer ) Making resource available\n(producer ) Notifying to all consumers\n(consumer2) Consumer consumed the resource\n(consumer1) Consumer consumed the resource\n"
    }
   ],
   "source": [
    "logging.basicConfig(level=logging.DEBUG,\n",
    "                    format='(%(threadName)-9s) %(message)s',)\n",
    "\n",
    "def consumer(cv):\n",
    "    logging.debug('Consumer thread started ...')\n",
    "    with cv:\n",
    "        logging.debug('Consumer waiting ...')\n",
    "        cv.wait()\n",
    "        logging.debug('Consumer consumed the resource')\n",
    "\n",
    "def producer(cv):\n",
    "    logging.debug('Producer thread started ...')\n",
    "    with cv:\n",
    "        logging.debug('Making resource available')\n",
    "        logging.debug('Notifying to all consumers')\n",
    "        cv.notifyAll()\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    condition = threading.Condition()\n",
    "    cs1 = threading.Thread(name='consumer1', target=consumer, args=(condition,))\n",
    "    cs2 = threading.Thread(name='consumer2', target=consumer, args=(condition,))\n",
    "    pd = threading.Thread(name='producer', target=producer, args=(condition,))\n",
    "\n",
    "    cs1.start()\n",
    "    cs2.start()\n",
    "    pd.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 多进程\n",
    "对应`multiprocessing`包，基础用法与`threading`相同，如下：注意jupyter中该用法有问题\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": "multil processing costs: 0.12417411804199219\n"
    }
   ],
   "source": [
    "import multiprocessing\n",
    "\n",
    "def mutil_processing():\n",
    "    process_list = []\n",
    "    for i in range(4):\n",
    "        p = multiprocessing.Process(target=f1, args=(i*num+1,))\n",
    "        p.start()\n",
    "        process_list.append(p)\n",
    "    for p in process_list:\n",
    "        p.join()\n",
    "    \n",
    "\n",
    "t3 = time.time()\n",
    "q3 = mutil_processing()\n",
    "print('multil processing costs:',time.time()-t3)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "也可以使用进程池来批量创建进程："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [],
   "source": [
    "def mutil_processing_pool():\n",
    "    pool=multiprocessing.Pool(4)\n",
    "    for i in range(4):\n",
    "        pool.apply_async(f1, args=(i*num+1,))\n",
    "    pool.close()\n",
    "    pool.join()\n",
    "    \n",
    "\n",
    "t4 = time.time()\n",
    "mutil_processing_pool()\n",
    "print('multil processing_pool costs:',time.time()-t4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ]
}